package cemp.rmq;

import cemp.common.constant.RedisKeyConstant;
import cemp.redis.util.RedisUtils;
import cemp.service.FetchDataService;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Resource;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.math.BigDecimal;
import java.util.Map;
import java.util.Optional;

@Component
public class StockKpiSZListener {
    @Autowired
    FetchDataService fetchDataService;
    @Resource
    private RedisUtils redisUtils;

//    @RabbitHandler
//    public void process(Map message) {
//
//        String  stockCode = (String)message.get("stockCode");
//        String date = (String)message.get("date");
//        //fetchDataService.staKpiRMQ(date,date,stockCode);
//        fetchDataService.staKpiLocal(date,stockCode);
//        System.out.println("success");
//    }


    @RabbitListener(queues = "topic.stock.kpi.sz")
    public void process(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        String stockCode = "";
        String date = "";
        try {
            ByteArrayInputStream strem = new ByteArrayInputStream(message.getBody());
            ObjectInputStream strem2 = new ObjectInputStream(strem);
            Map data = (Map) strem2.readObject();
            stockCode = (String) data.get("stockCode");
            date = (String)data.get("date");
            fetchDataService.staKpiLocal(date,stockCode);
        }catch (Exception e){
            handleDead(channel,tag,stockCode);
        }

    }
    private void handleDead(Channel channel, long tag,String stockCode) throws IOException {
        //异常消费 可以重复消费3次，超过了则进入死信队列等待特殊处理
        Optional obj = Optional.ofNullable(redisUtils.get(RedisKeyConstant.STOCK_KPI_OPS.concat(stockCode)));
        if (obj.isPresent()) {
            if (Integer.valueOf(obj.get().toString()) > 2) {
                channel.basicReject(tag, false);     //直接进死信 队列
            } else {
                redisUtils.set(RedisKeyConstant.STOCK_KPI_OPS.concat(stockCode), new BigDecimal(obj.get().toString()).add(BigDecimal.ONE).toString(), 3600);                   //处理次数+1
                channel.basicReject(tag, true);      //丢回队列 重新消费
            }
        } else {
            //redisUtils.set(RedisKeyConstant.STOCK_OPS.concat(stockCode),1,3600);
            String key = RedisKeyConstant.STOCK_KPI_OPS.concat(stockCode);
            redisUtils.set(key, "1", 3600);
            channel.basicReject(tag, true);      //丢回队列 重新消费
        }
    }
}
